package com.cloudfun.msglisten.listener;

import com.cloudfun.msglisten.common.constant.ExternalRabbitConstants;
import com.cloudfun.msglisten.common.constant.RabbitConstants;
import com.cloudfun.msglisten.common.model.dto.dingtalk.TextContent;
import com.cloudfun.msglisten.service.DingTalkService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * 用户开通数据订阅功能消息监听器
 * Created by Huqin on 2020/3/30
 */
@Slf4j
@Component
public class UserActivateDataSubMsgListener implements InitializingBean {

    @Autowired
    @Qualifier("externalRabbitAdmin")
    private RabbitAdmin rabbitAdmin;

    @Autowired
    private DingTalkService dingTalkService;

    private Map<String, DirectExchange> queueNameExchangeMap;

    private Map<String, Object> queueArguments;

    @Override
    public void afterPropertiesSet() throws Exception {
        // 对外的交换机
        DirectExchange statusExchange = new DirectExchange(ExternalRabbitConstants.Exchange.DEVICE_STATUS);
        DirectExchange faultExchange = new DirectExchange(ExternalRabbitConstants.Exchange.DEVICE_FAULT);
        DirectExchange eventExchange = new DirectExchange(ExternalRabbitConstants.Exchange.DEVICE_EVENT);
        rabbitAdmin.declareExchange(statusExchange);
        rabbitAdmin.declareExchange(faultExchange);
        rabbitAdmin.declareExchange(eventExchange);
        // 队列前缀和交换机
        queueNameExchangeMap = new HashMap<>();
        queueNameExchangeMap.put(ExternalRabbitConstants.Queue.DEVICE_STATUS_SUFFIX, statusExchange);
        queueNameExchangeMap.put(ExternalRabbitConstants.Queue.DEVICE_FAULT_SUFFIX, faultExchange);
        queueNameExchangeMap.put(ExternalRabbitConstants.Queue.DEVICE_EVENT_SUFFIX, eventExchange);

        // 队列参数
        queueArguments = new HashMap<>();
        // 统一设置队列中的所有消息的过期时间 24h
        queueArguments.put("x-message-ttl", 24 * 60 * 60 * 1000);
        // 设置超过多少毫秒没有消费者来访问队列，就删除队列的时间
        // queueArguments.put("x-expires", 20000);
        // 设置队列的最新的N条消息，如果超过N条，前面的消息将从队列中移除掉
        // queueArguments.put("x-max-length", 100_000);
        // 设置队列的内容的最大空间，超过该阈值就删除之前的消息
        // queueArguments.put("x-max-length-bytes", 1024);
        // 将删除的消息推送到指定的交换机，一般x-dead-letter-exchange和x-dead-letter-routing-key需要同时设置
        // queueArguments.put("x-dead-letter-exchange", "exchange.dead");
        // 将删除的消息推送到指定的交换机对应的路由键
        // queueArguments.put("x-dead-letter-routing-key", "routingkey.dead");
    }

    @RabbitHandler
    @RabbitListener(queues = RabbitConstants.Queue.USER_ACTIVATE_DATA_SUB_MSG_LISTEN)
    public void process(Map<String, String> map) {
        log.info("收到用户开通数据订阅功能消息：{}", map);
        try {
            String appId = map.get("appId");
            for (Map.Entry<String, DirectExchange> entry : queueNameExchangeMap.entrySet()) {
                String queueName = entry.getKey();
                DirectExchange exchange = entry.getValue();
                queueName = appId + "." + queueName;
                Queue queue = new Queue(queueName, true, false, false, queueArguments);
                rabbitAdmin.declareQueue(queue);
                Binding binding = BindingBuilder.bind(queue).to(exchange).with(appId);
                rabbitAdmin.declareBinding(binding);
            }
        } catch (Exception e) {
            log.error("收到用户开通数据订阅功能消息：{}处理异常：{}", map, e);
            TextContent textContent = new TextContent("app：" + map.get("appId") + "开通数据订阅功能异常：" + e);
            dingTalkService.sendMessage(textContent);
        }
    }

}
